#include "mqclient.h"
#include <QCoreApplication>
#include <QDateTime>
#include <QLoggingCategory>
#include <QThread>
#include <memory>



Q_LOGGING_CATEGORY(mqclient, "mqclient")

// ============================================================================
// MQClientPublishTask 实现
// ============================================================================

/**
 * @brief MQClientPublishTask构造函数
 * @param client MQClient实例指针
 * @param exchange 交换机名称
 * @param routingKey 路由键
 * @param message 消息内容
 * @param persistent 是否持久化
 */
MQClientPublishTask::MQClientPublishTask(MQClient* client, const QString& exchange, 
                                       const QString& routingKey, const QByteArray& message, 
                                       bool persistent)
    : m_client(client)
    , m_exchange(exchange)
    , m_routingKey(routingKey)
    , m_message(message)
    , m_persistent(persistent)
{
    // 设置任务自动删除
    setAutoDelete(true);
}

/**
 * @brief 执行异步发布任务
 */
void MQClientPublishTask::run()
{
    if (!m_client) {
        qCWarning(mqclient) << "MQClient实例为空，无法执行异步发布任务";
        return;
    }
    
    try {
        // 调用同步发布方法
        bool success = m_client->publishMessage(m_exchange, m_routingKey, m_message, m_persistent);
        
        if (success) {
            qCDebug(mqclient) << "异步发布消息成功:" << m_exchange << "/" << m_routingKey;
        } else {
            qCWarning(mqclient) << "异步发布消息失败:" << m_exchange << "/" << m_routingKey;
        }
        
        // 可以在这里发射信号通知结果（如果需要的话）
        // 注意：由于在工作线程中，信号需要通过Qt的信号槽机制传递到主线程
        
    } catch (const std::exception& e) {
        qCWarning(mqclient) << "异步发布消息时发生异常:" << e.what();
    } catch (...) {
        qCWarning(mqclient) << "异步发布消息时发生未知异常";
    }
}

// ============================================================================
// MQClient 实现
// ============================================================================

// 静态成员初始化
MQClient* MQClient::m_instance = nullptr;
QMutex MQClient::m_instanceMutex;

/**
 * @brief 获取MQClient单例实例
 * @return MQClient实例指针
 */
MQClient* MQClient::getInstance()
{
    if (m_instance == nullptr) {
        QMutexLocker locker(&m_instanceMutex);
        if (m_instance == nullptr) {
            m_instance = new MQClient();
        }
    }
    return m_instance;
}

/**
 * @brief 私有构造函数（单例模式）
 * @param parent 父对象
 */
MQClient::MQClient(QObject *parent) : QObject(parent),
    m_conn(nullptr),
    m_connected(0),
    m_connectionPool(nullptr),
    m_reconnectTimer(new QTimer(this)),
    m_reconnectAttempts(0),
    m_autoReconnectEnabled(false),
    m_heartbeatTimer(new QTimer(this)),
    m_publishedCount(0),
    m_consumedCount(0),
    m_errorCount(0)
{
    // 从配置管理器读取业务配置
    MQConfigManager* configManager = MQConfigManager::getInstance();
    m_businessConfig = configManager->getBusinessConfig();
    
    // 连接配置变更信号（使用队列连接避免竞态条件）
    QObject::connect(configManager, &MQConfigManager::configChanged,
                     this, &MQClient::onConfigChanged, Qt::QueuedConnection);
    
    //GJ  todo是否需要处理，对话框告警？
    // 如果配置为空，则使用默认值
//    if (m_businessConfig.defaultExchange.name.isEmpty()) {
//        m_businessConfig.defaultExchange.name = "agv.exchange";
//        m_businessConfig.defaultExchange.type = "direct";
//        m_businessConfig.defaultExchange.durable = true;
        
//        m_businessConfig.defaultQueue.name = "agv.queue";
//        m_businessConfig.defaultQueue.durable = true;
//        m_businessConfig.defaultQueue.exclusive = false;
//        m_businessConfig.defaultQueue.autoDelete = false;
        
//        m_businessConfig.defaultBinding.routingKey = "agv.routing.key";
//    }
    
    // 设置重连定时器
    m_reconnectTimer->setSingleShot(true);
    QObject::connect(m_reconnectTimer, &QTimer::timeout,
                     this, &MQClient::onReconnectTimer);
    
    // 设置心跳定时器
    QObject::connect(m_heartbeatTimer, &QTimer::timeout,
                     this, &MQClient::onHeartbeatTimer);
    
    // 注意：异步处理现在使用QThreadPool，不再需要专用的异步工作线程
    
    qCDebug(mqclient) << "MQClient单例实例已创建";
}

/**
 * @brief 析构函数
 */
MQClient::~MQClient()
{
    disconnect();
    
    // 等待所有异步任务完成
    QThreadPool::globalInstance()->waitForDone(3000);
    
    // 清理连接池
    if (m_connectionPool) {
        m_connectionPool->cleanup();
        m_connectionPool.reset();
    }
    
    qCDebug(mqclient) << "MQClient实例已销毁";
}

/**
 * @brief 使用配置文件连接到RabbitMQ服务器
 * @return 是否连接成功
 */
bool MQClient::connect()
{
    // 从配置管理器获取连接配置
    MQConfigManager* configManager = MQConfigManager::getInstance();
    MQConnectionConfig config = configManager->getConnectionConfig();
    
    return connectInternal(config);
}



/**
 * @brief 内部连接实现
 * @param config 连接配置
 * @return 是否连接成功
 */
bool MQClient::connectInternal(const MQConnectionConfig& config)
{
    QMutexLocker locker(&m_connectionMutex);
    // 如果已经连接，先断开
    if (m_connected.load()) {
        disconnect();
    }
    
    qCDebug(mqclient) << "开始连接到" << config.hostname << ":" << config.port;
    
    // 创建连接状态
    m_conn = amqp_new_connection();
    if (!m_conn) {
        m_lastError = "无法创建RabbitMQ连接";
        emit error(m_lastError);
        m_errorCount.fetchAndAddOrdered(1);
        return false;
    }
    
    // 创建套接字
    amqp_socket_t *socket = amqp_tcp_socket_new(m_conn);
    if (!socket) {
        m_lastError = "无法创建TCP套接字";
        emit error(m_lastError);
        amqp_destroy_connection(m_conn);
        m_conn = nullptr;
        m_errorCount.fetchAndAddOrdered(1);
        return false;
    }
    
    // 打开套接字连接
    int status = amqp_socket_open(socket, config.hostname.toUtf8().constData(), 
                                 config.port);
    if (status != AMQP_STATUS_OK) {
        m_lastError = QString("无法打开TCP套接字连接: %1").arg(status);
        emit error(m_lastError);
        amqp_destroy_connection(m_conn);
        m_conn = nullptr;
        m_errorCount.fetchAndAddOrdered(1);
        return false;
    }
    
    // 登录到RabbitMQ服务器
    amqp_rpc_reply_t reply = amqp_login(m_conn, config.vhost.toUtf8().constData(),
                                       0, 131072, 60,  // 使用默认值：frame_max=131072, heartbeat=60
                                       AMQP_SASL_METHOD_PLAIN,
                                       config.username.toUtf8().constData(),
                                       config.password.toUtf8().constData());
    if (!checkRpcReply(reply, "登录")) {
        amqp_destroy_connection(m_conn);
        m_conn = nullptr;
        m_errorCount.fetchAndAddOrdered(1);
        return false;
    }
    
    // 打开通道
    amqp_channel_open(m_conn, 1);
    reply = amqp_get_rpc_reply(m_conn);
    if (!checkRpcReply(reply, "打开通道")) {
        amqp_destroy_connection(m_conn);
        m_conn = nullptr;
        m_errorCount.fetchAndAddOrdered(1);
        return false;
    }
    
    m_connected.store(1);
    m_reconnectAttempts.store(0);
    
    // 连接池管理逻辑
    const int CONNECTION_POOL_SIZE = 5;  // 连接池大小
    
    // 检查并清理现有连接池
    if (m_connectionPool) {
        qCDebug(mqclient) << "检测到现有连接池，进行清理和重建";
        
        // 安全清理：先停止定时器，再清理连接
        m_connectionPool->cleanup();
        m_connectionPool.reset();
        
        qCDebug(mqclient) << "旧连接池已清理";
    } else {
        qCDebug(mqclient) << "首次创建连接池";
    }
    
    // 创建新连接池
    m_connectionPool = std::make_unique<MQConnectionPool>(config, CONNECTION_POOL_SIZE, this);
    if (!m_connectionPool->initialize(config)) {
        qCWarning(mqclient) << "连接池初始化失败";
        // 清理资源
        amqp_destroy_connection(m_conn);
        m_conn = nullptr;
        m_connected.store(0);
        m_errorCount.fetchAndAddOrdered(1);
        return false;
    }
    
    qCInfo(mqclient) << "连接池初始化完成";
    
    // 启动心跳检测（硬编码配置）
    const bool ENABLE_HEARTBEAT = true;  // 启用心跳检测
    const int HEARTBEAT_INTERVAL_MS = 5000;  // 心跳检测间隔5秒
    if (ENABLE_HEARTBEAT) {
        m_heartbeatTimer->start(HEARTBEAT_INTERVAL_MS);
    }
    
    emit connectionChanged(true);
    qCDebug(mqclient) << "连接成功";
    return true;
}

/**
 * @brief 断开连接
 */
void MQClient::disconnect()
{
    QMutexLocker locker(&m_connectionMutex);
    
    if (m_conn) {
        if (m_connected.load()) {
            qCDebug(mqclient) << "开始断开连接";
            
            // 停止心跳检测
            m_heartbeatTimer->stop();
            

            // 关闭通道
            amqp_channel_close(m_conn, 1, AMQP_REPLY_SUCCESS);
            // 关闭连接
            amqp_connection_close(m_conn, AMQP_REPLY_SUCCESS);
        }
        // 销毁连接
        amqp_destroy_connection(m_conn);
        m_conn = nullptr;
        m_connected.store(0);
        emit connectionChanged(false);
        qCDebug(mqclient) << "连接已断开";
    }
}

/**
 * @brief 启动自动重连
 */
void MQClient::startAutoReconnect()
{
    m_autoReconnectEnabled = true;
    qCDebug(mqclient) << "自动重连已启用";
}

/**
 * @brief 停止自动重连
 */
void MQClient::stopAutoReconnect()
{
    m_autoReconnectEnabled = false;
    m_reconnectTimer->stop();
    qCDebug(mqclient) << "自动重连已禁用";
}

/**
 * @brief 执行重连
 */
void MQClient::performReconnect()
{
    if (!m_autoReconnectEnabled) {
        return;
    }
    
    // 硬编码的可靠性配置
    const int MAX_RECONNECT_ATTEMPTS = 10;  // 最大重连尝试次数
    const int RECONNECT_DELAY_MS = 5000;    // 重连延迟5秒
    
    int currentAttempts = m_reconnectAttempts.load();
    
    if (currentAttempts >= MAX_RECONNECT_ATTEMPTS) {
        qCWarning(mqclient) << "重连尝试次数已达上限:" << MAX_RECONNECT_ATTEMPTS;
        emit reconnectFailed(currentAttempts);
        return;
    }
    
    qCDebug(mqclient) << "开始第" << (currentAttempts + 1) << "次重连尝试";
    
    // 修改：使用connectAndInitialize()替代connect()
    if (connectAndInitialize()) {
        qCInfo(mqclient) << "重连成功";
        emit reconnected();
    } else {
        m_reconnectAttempts.fetchAndAddOrdered(1);
        qCWarning(mqclient) << "重连失败，将在" << RECONNECT_DELAY_MS << "ms后重试";
        m_reconnectTimer->start(RECONNECT_DELAY_MS);
    }
}

/**
 * @brief 心跳检测
 */
/**
 * @brief 执行心跳检测
 * 通过尝试执行一个轻量级的AMQP操作来检测连接状态
 */
void MQClient::performHeartbeat()
{
    bool needReconnect = false;
    
    {
        QMutexLocker locker(&m_connectionMutex);
        
        if (!m_connected.load() || !m_conn) {
            emit heartbeatCheck(false);
            return;
        }
        
        // 尝试执行一个轻量级操作来检测连接状态
        // 使用 amqp_channel_flow 来测试连接是否仍然有效
        amqp_channel_flow(m_conn, 1, 1);  // 通道1，启用流控制
        amqp_rpc_reply_t reply = amqp_get_rpc_reply(m_conn);
        
        bool alive = (reply.reply_type == AMQP_RESPONSE_NORMAL);
        
        emit heartbeatCheck(alive);
        
        if (!alive) {
            qCWarning(mqclient) << "心跳检测失败，连接已断开，错误类型:" << reply.reply_type;
            m_connected.store(0);
            emit connectionChanged(false);
            
            // 标记需要重连，但不在锁内执行重连
            needReconnect = m_autoReconnectEnabled;
        } else {
            //GJ 正常暂时不打印
            //qCDebug(mqclient) << "心跳检测成功，连接正常";
        }
    } // 锁在这里释放
    
    // 在锁外执行重连，避免死锁
    if (needReconnect) {
        performReconnect();
    }
}

/**
 * @brief 重连定时器超时处理
 */
void MQClient::onReconnectTimer()
{
    performReconnect();
}

/**
 * @brief 心跳定时器超时处理
 */
void MQClient::onHeartbeatTimer()
{
    performHeartbeat();
}

/**
 * @brief 配置变更槽函数
 * 当配置管理器发出配置变更信号时，重新加载业务配置
 */
void MQClient::onConfigChanged()
{
    qCDebug(mqclient) << "配置已更新，重新加载业务配置";
    
    // 保存当前的业务配置用于解绑
    MQBusinessConfig oldConfig = m_businessConfig;
    
    // 从配置管理器重新读取业务配置
    MQConfigManager* configManager = MQConfigManager::getInstance();
    if (configManager) {
        MQBusinessConfig newConfig = configManager->getBusinessConfig();
        
        // 如果配置为空，使用默认配置
        if (newConfig.defaultExchange.name.isEmpty()) {
            qCDebug(mqclient) << "配置为空，使用默认业务配置";
            
            // 设置默认交换机
            newConfig.defaultExchange.name = "agv.exchange";
            newConfig.defaultExchange.type = "direct";
            newConfig.defaultExchange.durable = true;
            
            // 设置默认队列
            newConfig.defaultQueue.name = "agv.queue";
            newConfig.defaultQueue.durable = true;
            newConfig.defaultQueue.exclusive = false;
            newConfig.defaultQueue.autoDelete = false;
            
            // 设置默认绑定
            newConfig.defaultBinding.routingKey = "agv.routing.key";
            newConfig.defaultBinding.queue_routing_key = "agv.routing.key";
        }
        
        // 如果连接中且配置发生变化，先解绑旧的绑定
        if (m_connected.load() && !oldConfig.defaultQueue.name.isEmpty() && 
            !oldConfig.defaultExchange.name.isEmpty() && 
            !oldConfig.defaultBinding.queue_routing_key.isEmpty()) {
            
            // 检查绑定配置是否发生变化
            bool bindingChanged = (oldConfig.defaultQueue.name != newConfig.defaultQueue.name) ||
                                (oldConfig.defaultExchange.name != newConfig.defaultExchange.name) ||
                                (oldConfig.defaultBinding.queue_routing_key != newConfig.defaultBinding.queue_routing_key);
            
            if (bindingChanged) {
                qCDebug(mqclient) << "绑定配置发生变化，解绑旧的绑定";
                qCDebug(mqclient) << "解绑:" << oldConfig.defaultQueue.name 
                                 << "从" << oldConfig.defaultExchange.name 
                                 << "路由键:" << oldConfig.defaultBinding.queue_routing_key;
                
                // 解绑旧的绑定
                if (!unbindQueue(oldConfig.defaultQueue.name, 
                               oldConfig.defaultExchange.name, 
                               oldConfig.defaultBinding.queue_routing_key)) {
                    qCWarning(mqclient) << "解绑旧绑定失败，但继续设置新配置";
                }
            }
        }
        
        // 更新业务配置
        setBusinessConfig(newConfig);
        qCInfo(mqclient) << "业务配置已更新";
    }
    
    // 如果连接中，重新应用心跳配置（硬编码配置）
    if (m_connected.load()) {
        const bool ENABLE_HEARTBEAT = true;  // 启用心跳检测
        const int HEARTBEAT_INTERVAL_MS = 30000;  // 心跳检测间隔30秒
        if (ENABLE_HEARTBEAT) {
            m_heartbeatTimer->start(HEARTBEAT_INTERVAL_MS);
        } else {
            m_heartbeatTimer->stop();
        }
    }
}

/**
 * @brief 声明交换机
 * @param exchange 交换机名称
 * @param type 交换机类型 (direct, fanout, topic, headers)
 * @param durable 是否持久化
 * @return 是否成功
 */
bool MQClient::declareExchange(const QString &exchange, const QString &type, bool durable)
{
    QMutexLocker locker(&m_connectionMutex);
    
    if (!m_connected.load() || !m_conn) {
        m_lastError = "未连接到RabbitMQ服务器";
        emit error(m_lastError);
        return false;
    }
    
    // 声明交换机
    amqp_exchange_declare(m_conn, 1,
                          amqp_cstring_bytes(exchange.toUtf8().constData()),
                          amqp_cstring_bytes(type.toUtf8().constData()),
                          0, // passive
                          durable ? 1 : 0, // durable
                          0, // auto_delete
                          0, // internal
                          amqp_empty_table);
    
    amqp_rpc_reply_t reply = amqp_get_rpc_reply(m_conn);
    return checkRpcReply(reply, "声明交换机");
}

/**
 * @brief 声明队列
 * @param queue 队列名称
 * @param durable 是否持久化
 * @param exclusive 是否排他
 * @param autoDelete 是否自动删除
 * @return 是否成功
 */
bool MQClient::declareQueue(const QString &queue, bool durable, bool exclusive, bool autoDelete)
{
    QMutexLocker locker(&m_connectionMutex);
    
    if (!m_connected.load() || !m_conn) {
        m_lastError = "未连接到RabbitMQ服务器";
        emit error(m_lastError);
        return false;
    }
    
    // 声明队列
    amqp_queue_declare(m_conn, 1,
                       amqp_cstring_bytes(queue.toUtf8().constData()),
                       0, // passive
                       durable ? 1 : 0, // durable
                       exclusive ? 1 : 0, // exclusive
                       autoDelete ? 1 : 0, // auto_delete
                       amqp_empty_table);
    
    amqp_rpc_reply_t reply = amqp_get_rpc_reply(m_conn);
    return checkRpcReply(reply, "声明队列");
}

/**
 * @brief 绑定队列到交换机
 * @param queue 队列名称
 * @param exchange 交换机名称
 * @param routingKey 路由键
 * @return 是否成功
 */
bool MQClient::bindQueue(const QString &queue, const QString &exchange, const QString &routingKey)
{
    QMutexLocker locker(&m_connectionMutex);
    
    if (!m_connected.load() || !m_conn) {
        m_lastError = "未连接到RabbitMQ服务器";
        emit error(m_lastError);
        return false;
    }
    
    // 绑定队列到交换机
    amqp_queue_bind(m_conn, 1,
                    amqp_cstring_bytes(queue.toUtf8().constData()),
                    amqp_cstring_bytes(exchange.toUtf8().constData()),
                    amqp_cstring_bytes(routingKey.toUtf8().constData()),
                    amqp_empty_table);
    
    amqp_rpc_reply_t reply = amqp_get_rpc_reply(m_conn);
    return checkRpcReply(reply, "绑定队列");
}

/**
 * @brief 解绑队列与交换机的绑定
 * @param queue 队列名称
 * @param exchange 交换机名称
 * @param routingKey 路由键
 * @return 是否成功
 */
bool MQClient::unbindQueue(const QString &queue, const QString &exchange, const QString &routingKey)
{
    QMutexLocker locker(&m_connectionMutex);
    
    if (!m_connected.load() || !m_conn) {
        m_lastError = "未连接到RabbitMQ服务器";
        emit error(m_lastError);
        return false;
    }
    
    // 解绑队列与交换机的绑定
    amqp_queue_unbind(m_conn, 1,
                      amqp_cstring_bytes(queue.toUtf8().constData()),
                      amqp_cstring_bytes(exchange.toUtf8().constData()),
                      amqp_cstring_bytes(routingKey.toUtf8().constData()),
                      amqp_empty_table);
    
    amqp_rpc_reply_t reply = amqp_get_rpc_reply(m_conn);
    return checkRpcReply(reply, "解绑队列");
}

/**
 * @brief 发布消息到指定交换机和路由键（使用连接池）
 * @param exchange 交换机名称
 * @param routingKey 路由键
 * @param message 消息内容
 * @param persistent 是否持久化
 * @return 是否成功
 */
bool MQClient::publishMessage(const QString &exchange, const QString &routingKey, const QByteArray &message, bool persistent)
{
    if (!m_connected.load()) {
        m_lastError = "未连接到RabbitMQ服务器";
        emit error(m_lastError);
        m_errorCount++;
        return false;
    }
    
    if (!m_connectionPool) {
        m_lastError = "连接池未初始化";
        emit error(m_lastError);
        m_errorCount++;
        return false;
    }
    
    // 从连接池获取连接（使用非阻塞方式，避免UI线程阻塞）
    auto connection = m_connectionPool->getConnection(0);  // 0表示不等待，立即返回
    if (!connection) {
        m_lastError = "无法从连接池获取连接（连接池繁忙或已满）";
        emit error(m_lastError);
        m_errorCount++;
        return false;
    }
    
    bool result = false;
    
    try {
        // 设置消息属性
        amqp_basic_properties_t props;
        props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
        props.content_type = amqp_cstring_bytes("text/plain");
        props.delivery_mode = persistent ? 2 : 1; // 2表示持久化，1表示非持久化
        
        // 创建消息体（修复内存泄漏）
        amqp_bytes_t messageBody;
        messageBody.len = message.size();
        amqp_bytes_t allocatedBytes = amqp_bytes_malloc(messageBody.len);
        messageBody.bytes = allocatedBytes.bytes;
        if (!messageBody.bytes) {
            m_lastError = "内存分配失败";
            emit error(m_lastError);
            m_errorCount++;
            // 归还连接并返回失败
            m_connectionPool->returnConnection(connection);
            return false;
        }
        memcpy(messageBody.bytes, message.constData(), messageBody.len);
        
        // 发布消息
        int status = amqp_basic_publish(connection, 1,
                                       amqp_cstring_bytes(exchange.toUtf8().constData()),
                                       amqp_cstring_bytes(routingKey.toUtf8().constData()),
                                       0, // mandatory
                                       0, // immediate
                                       &props,
                                       messageBody);
        
        // 释放内存（修复内存泄漏）
        amqp_bytes_free(messageBody);
        
        if (status != AMQP_STATUS_OK) {
            qCDebug(mqclient) << "消息发布失败:" << exchange << "/" << routingKey;
            m_lastError = QString("发布消息失败: %1").arg(status);
            emit error(m_lastError);
            m_errorCount++;
            result = false;
        } else {
            m_publishedCount++;
            qCDebug(mqclient) << "消息发布成功:" << exchange << "/" << routingKey;
            result = true;
        }
        
    } catch (const std::exception& e) {
        m_lastError = QString("发布消息时发生异常: %1").arg(e.what());
        emit error(m_lastError);
        m_errorCount++;
        result = false;
    } catch (...) {
        m_lastError = "发布消息时发生未知异常";
        emit error(m_lastError);
        m_errorCount++;
        result = false;
    }
    
    // 归还连接到连接池
    m_connectionPool->returnConnection(connection);
    
    return result;
}

/**
 * @brief 使用默认业务配置发布消息（简化接口）
 * @param message 消息内容
 * @param persistent 是否持久化消息
 * @return 是否发布成功
 */
bool MQClient::publishMessage(const QByteArray &message, bool persistent)
{
    // 获取默认业务配置
    QMutexLocker configLocker(&m_businessConfigMutex);
    QString defaultExchange = m_businessConfig.defaultExchange.name;
    QString defaultRoutingKey = m_businessConfig.defaultBinding.routingKey;
    configLocker.unlock();
    
    // 检查配置是否有效
    if (defaultExchange.isEmpty() || defaultRoutingKey.isEmpty()) {
        m_lastError = "默认业务配置无效：交换机或路由键为空";
        emit error(m_lastError);
        m_errorCount++;
        return false;
    }
    
    // 调用原有的发布方法
    return publishMessage(defaultExchange, defaultRoutingKey, message, persistent);
}



/**
 * @brief 异步发布消息
 * @param exchange 交换机名称
 * @param routingKey 路由键
 * @param message 消息内容
 * @param persistent 是否持久化
 * @return 是否成功提交到线程池
 */
bool MQClient::publishMessageAsync(const QString &exchange, const QString &routingKey, const QByteArray &message, bool persistent)
{
    try {
        // 创建异步发布任务
        MQClientPublishTask* task = new MQClientPublishTask(this, exchange, routingKey, message, persistent);
        
        // 提交到全局线程池
        QThreadPool::globalInstance()->start(task);
        
        qCDebug(mqclient) << "消息已提交到异步线程池:" << exchange << "/" << routingKey;
        return true;
        
    } catch (const std::exception& e) {
        qCWarning(mqclient) << "提交异步发布任务时发生异常:" << e.what();
        return false;
    } catch (...) {
        qCWarning(mqclient) << "提交异步发布任务时发生未知异常";
        return false;
    }
}

/**
 * @brief 使用默认业务配置异步发布消息（简化接口）
 * @param message 消息内容
 * @param persistent 是否持久化消息
 * @return 是否成功提交到线程池
 */
bool MQClient::publishMessageAsync(const QByteArray &message, bool persistent)
{
    // 获取默认业务配置
    QMutexLocker configLocker(&m_businessConfigMutex);
    QString defaultExchange = m_businessConfig.defaultExchange.name;
    QString defaultRoutingKey = m_businessConfig.defaultBinding.routingKey;
    configLocker.unlock();
    
    // 检查配置是否有效
    if (defaultExchange.isEmpty() || defaultRoutingKey.isEmpty()) {
        m_lastError = "默认业务配置无效：交换机或路由键为空";
        emit error(m_lastError);
        m_errorCount++;
        return false;
    }
    
    // 调用完整的异步发布方法
    return publishMessageAsync(defaultExchange, defaultRoutingKey, message, persistent);
}



/**
 * @brief 消费消息（同步方式）
 * @param queue 队列名称
 * @param autoAck 是否自动确认
 * @return 消息内容，如果没有消息则返回空字符串
 */
QByteArray MQClient::consumeMessage(const QString &queue, bool autoAck)
{
    if (!m_connected.load()) {
        m_lastError = "未连接到RabbitMQ服务器";
        emit error(m_lastError);
        m_errorCount++;
        return QByteArray();
    }

    if (!m_connectionPool) {
        m_lastError = "连接池未初始化";
        emit error(m_lastError);
        m_errorCount++;
        return QByteArray();
    }

    // 从连接池获取连接（使用非阻塞方式，避免UI线程阻塞）
    auto connection = m_connectionPool->getConnection(0);  // 0表示不等待，立即返回
    if (!connection) {
        m_lastError = "无法从连接池获取连接（连接池繁忙或已满）";
        emit error(m_lastError);
        m_errorCount++;
        return QByteArray();
    }

    QByteArray result;
    
    try {
        // 获取消息
        amqp_rpc_reply_t reply;
        amqp_message_t message;
        
        amqp_maybe_release_buffers(connection);
        
        // 获取消息
        reply = amqp_basic_get(connection, 1,
                               amqp_cstring_bytes(queue.toUtf8().constData()),
                               autoAck ? 1 : 0);
        
        if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
            // 先检查是否是队列为空的情况（这是正常情况）
            if (reply.reply.id == AMQP_BASIC_GET_EMPTY_METHOD) {
                // 队列为空，这是正常情况
                qCDebug(mqclient) << "队列为空:" << queue;
                result = QByteArray();
            } else {
                // 其他错误情况，使用checkRpcReply获取详细错误信息
                if (!checkRpcReply(reply, QString("消费消息[%1]").arg(queue))) {
                    // checkRpcReply已经设置了m_lastError和发射了error信号
                    m_errorCount++;
                    // 归还连接并返回空结果
                    m_connectionPool->returnConnection(connection);
                    return QByteArray();
                }
            }
        } else {
            // reply.reply_type == AMQP_RESPONSE_NORMAL，但需要检查是否真的有消息
            // 检查reply.reply.id是否表示有消息
            if (reply.reply.id == AMQP_BASIC_GET_OK_METHOD) {
                // 有消息，读取消息体
                amqp_read_message(connection, 1, &message, 0);
                
                // 转换消息内容为QByteArray
                result = QByteArray((const char*)message.body.bytes, message.body.len);
                
                // 释放消息
                amqp_destroy_message(&message);
                
                m_consumedCount++;
                qCDebug(mqclient) << "消息消费成功:" << queue << "内容长度:" << result.size();
            } else {               
                // AMQP_RESPONSE_NORMAL但不是AMQP_BASIC_GET_OK_METHOD，可能是其他响应
                //GJ 队列无消息时，属于正常情况，暂时不打印
//                qCDebug(mqclient) << "收到意外的响应类型:" << reply.reply.id << "队列:" << queue;
                result = QByteArray();
            }
        }
    } catch (const std::exception& e) {
        m_lastError = QString("消费消息时发生异常: %1").arg(e.what());
        emit error(m_lastError);
        m_errorCount++;
    } catch (...) {
        m_lastError = "消费消息时发生未知异常";
        emit error(m_lastError);
        m_errorCount++;
    }
    
    // 归还连接到连接池
    m_connectionPool->returnConnection(connection);
    
    return result;
}

/**
 * @brief 使用默认业务配置从队列消费消息（简化接口）
 * @param autoAck 是否自动确认
 * @return 消息内容，如果没有消息则返回空
 */
QByteArray MQClient::consumeMessage(bool autoAck)
{
    // 获取默认业务配置
    QMutexLocker configLocker(&m_businessConfigMutex);
    QString defaultQueue = m_businessConfig.defaultQueue.name;
    configLocker.unlock();
    
    // 检查配置是否有效
    if (defaultQueue.isEmpty()) {
        m_lastError = "默认业务配置无效：队列名称为空";
        emit error(m_lastError);
        m_errorCount++;
        return QByteArray();
    }
    
    // 调用原有的消费方法
    return consumeMessage(defaultQueue, autoAck);
}

/**
 * @brief 确认消息
 * @param deliveryTag 投递标签
 * @return 是否成功
 */
bool MQClient::ackMessage(uint64_t deliveryTag)
{
    if (!m_connected.load()) {
        m_lastError = "未连接到RabbitMQ服务器";
        emit error(m_lastError);
        m_errorCount++;
        return false;
    }
    
    if (!m_connectionPool) {
        m_lastError = "连接池未初始化";
        emit error(m_lastError);
        m_errorCount++;
        return false;
    }
    
    // 从连接池获取连接
    auto connection = m_connectionPool->getConnection();
    if (!connection) {
        m_lastError = "无法从连接池获取连接";
        emit error(m_lastError);
        m_errorCount++;
        return false;
    }
    
    bool result = false;
    
    try {
        // 确认消息
        int status = amqp_basic_ack(connection, 1, deliveryTag, 0);
        
        if (status != AMQP_STATUS_OK) {
            m_lastError = QString("确认消息失败: %1").arg(status);
            emit error(m_lastError);
            m_errorCount++;
            result = false;
        } else {
            qCDebug(mqclient) << "消息确认成功:" << deliveryTag;
            result = true;
        }
    } catch (const std::exception& e) {
        m_lastError = QString("确认消息时发生异常: %1").arg(e.what());
        emit error(m_lastError);
        m_errorCount++;
        result = false;
    } catch (...) {
        m_lastError = "确认消息时发生未知异常";
        emit error(m_lastError);
        m_errorCount++;
        result = false;
    }
    
    // 归还连接到连接池
    m_connectionPool->returnConnection(connection);
    
    return result;
}

/**
 * @brief 是否已连接
 * @return 连接状态
 */
bool MQClient::isConnected() const
{
    return m_connected.load();
}

/**
 * @brief 获取连接池状态
 * @return 连接池状态信息
 */
QString MQClient::getConnectionPoolStatus() const
{
    if (!m_connectionPool) {
        return "连接池未初始化";
    }
    return m_connectionPool->getStatus();
}

// 注意：getAsyncQueueStatus方法已移除，因为现在使用QThreadPool进行异步处理

/**
 * @brief 获取最后的错误信息
 * @return 错误信息
 */
QString MQClient::getLastError() const
{
    return m_lastError;
}

// 注意：clearAsyncQueue方法已移除，因为现在使用QThreadPool进行异步处理

/**
 * @brief 初始化默认拓扑结构（交换机、队列、绑定）
 * @return 是否初始化成功
 */
bool MQClient::initializeDefaultTopology()
{
    if (!isConnected()) {
        m_lastError = "未连接到MQ服务器";
        qCWarning(mqclient) << m_lastError;
        return false;
    }
    
    // 获取业务配置
    QMutexLocker locker(&m_businessConfigMutex);
    MQBusinessConfig businessConfig = m_businessConfig;
    
    // 声明默认交换机
    if (!declareExchange(businessConfig.defaultExchange.name, 
                        businessConfig.defaultExchange.type, 
                        businessConfig.defaultExchange.durable)) {
        m_lastError = QString("声明交换机失败: %1").arg(businessConfig.defaultExchange.name);
        qCWarning(mqclient) << m_lastError;
        return false;
    }
    qCDebug(mqclient) << "交换机声明成功:" << businessConfig.defaultExchange.name;
    
    // 声明默认队列
    if (!declareQueue(businessConfig.defaultQueue.name,
                     businessConfig.defaultQueue.durable,
                     businessConfig.defaultQueue.exclusive,
                     businessConfig.defaultQueue.autoDelete)) {
        m_lastError = QString("声明队列失败: %1").arg(businessConfig.defaultQueue.name);
        qCWarning(mqclient) << m_lastError;
        return false;
    }
    qCDebug(mqclient) << "队列声明成功:" << businessConfig.defaultQueue.name;
    
    // 绑定队列到交换机（使用队列绑定路由键）
    if (!bindQueue(businessConfig.defaultQueue.name,
                  businessConfig.defaultExchange.name,
                  businessConfig.defaultBinding.queue_routing_key)) {
        m_lastError = QString("队列绑定失败: %1 -> %2")
                     .arg(businessConfig.defaultQueue.name)
                     .arg(businessConfig.defaultExchange.name);
        qCWarning(mqclient) << m_lastError;
        return false;
    }
    qCDebug(mqclient) << "队列绑定成功:" << businessConfig.defaultQueue.name 
                     << "->" << businessConfig.defaultExchange.name 
                     << "(" << businessConfig.defaultBinding.queue_routing_key << ")";
    
    qCInfo(mqclient) << "默认拓扑结构初始化成功";
    return true;
}

/**
 * @brief 设置业务配置（用于配置界面）
 * @param config 业务配置
 */
void MQClient::setBusinessConfig(const MQBusinessConfig& config)
{
    QMutexLocker locker(&m_businessConfigMutex);
    m_businessConfig = config;
    qCDebug(mqclient) << "业务配置已更新";
}

/**
 * @brief 获取当前业务配置
 * @return 业务配置
 */
MQBusinessConfig MQClient::getBusinessConfig() const
{
    QMutexLocker locker(&m_businessConfigMutex);
    return m_businessConfig;
}

/**
 * @brief 连接并初始化默认拓扑结构的便捷方法
 * @return 是否连接并初始化成功
 */
bool MQClient::connectAndInitialize()
{
    // 先尝试连接
    if (!connect()) {
        return false;
    }
    
    // 连接成功后初始化拓扑结构
    if (!initializeDefaultTopology()) {
        return false;
    }
    
    // 启动自动重连
    startAutoReconnect();
    
    qCInfo(mqclient) << "MQ连接并初始化完成";
    return true;
}

/**
 * @brief 检查RabbitMQ操作的返回值
 * @param x 返回值
 * @param context 上下文信息
 * @return 是否成功
 */
bool MQClient::checkRpcReply(amqp_rpc_reply_t x, const QString &context)
{
    switch (x.reply_type) {
        case AMQP_RESPONSE_NORMAL:
            return true;
            
        case AMQP_RESPONSE_NONE:
            m_lastError = QString("%1: 异常：无响应").arg(context);
            break;
            
        case AMQP_RESPONSE_LIBRARY_EXCEPTION:
            m_lastError = QString("%1: 库异常: %2").arg(context).arg(amqp_error_string2(x.library_error));
            break;
            
        case AMQP_RESPONSE_SERVER_EXCEPTION:
            switch (x.reply.id) {
                case AMQP_CONNECTION_CLOSE_METHOD: {
                    amqp_connection_close_t *m = (amqp_connection_close_t *) x.reply.decoded;
                    m_lastError = QString("%1: 服务器连接关闭: %2, 消息: %3")
                                  .arg(context)
                                  .arg(m->reply_code)
                                  .arg(QString::fromUtf8((char*)m->reply_text.bytes, m->reply_text.len));
                    break;
                }
                case AMQP_CHANNEL_CLOSE_METHOD: {
                    amqp_channel_close_t *m = (amqp_channel_close_t *) x.reply.decoded;
                    m_lastError = QString("%1: 服务器通道关闭: %2, 消息: %3")
                                  .arg(context)
                                  .arg(m->reply_code)
                                  .arg(QString::fromUtf8((char*)m->reply_text.bytes, m->reply_text.len));
                    break;
                }
                default:
                    m_lastError = QString("%1: 未知服务器异常").arg(context);
                    break;
            }
            break;
            
        default:
            m_lastError = QString("%1: 未知错误").arg(context);
            break;
    }
    
    emit error(m_lastError);
    return false;
}
